package com.beiding.dbrequest.client;

import com.beiding.dbrequest.common.Message;
import io.netty.channel.socket.SocketChannel;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import lombok.Getter;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 客户端组
 * 多通道,提供负载能力
 */
public class ClientGroup {

    /**
     * +
     * 实例数
     */
    @Getter
    private int clientNumber;

    private List<Client> clients;

    public ClientGroup(int clientNumber) {
        if (clientNumber <= 0) {
            throw new IllegalArgumentException("客户端数量必须大于0");
        }
        this.clientNumber = clientNumber;
        clients = new ArrayList<>();
        for (int i = 0; i <= clientNumber; i++) {
            clients.add(new Client());
        }
    }


    /**
     * 随机获取一个实例
     *
     * @return
     */
    public Client getOne() {
        return clients.get(new Random().nextInt(clients.size()));
    }

    public Observable<Serializable> request(Message message) {
        Client one = getOne();
        return one.request(message);
    }

    public static ClientGroup create(int number, String host, Integer port) {
        ClientGroup clientGroup = new ClientGroup(number);
        clientGroup.setHost(host);
        clientGroup.setPort(port);
        return clientGroup;
    }


    public void setHost(String v) {
        for (Client client : clients) {
            client.setTargetHost(v);
        }
    }

    public void setPort(Integer v) {
        for (Client client : clients) {
            client.setTargetPort(v);
        }
    }

    public List<Client> getClients() {
        return Collections.unmodifiableList(clients);
    }

    public Observable<Client> start() {
        return Observable.create(subscriber -> {

            AtomicInteger atomicInteger = new AtomicInteger(0);

            int size = clients.size();
            for (Client client : clients) {
                client.start().subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable disposable) {

                    }

                    @Override
                    public void onComplete() {

                        subscriber.onNext(client);

                        if (atomicInteger.addAndGet(1) == size) {
                            subscriber.onComplete();
                        }
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        subscriber.onError(throwable);
                    }

                    @Override
                    public void onNext(Object o) {

                    }
                });
            }
        });
    }

    public Observable<Client> close() {
        return Observable.create(subscriber -> {

            AtomicInteger atomicInteger = new AtomicInteger(0);

            int size = clients.size();
            for (Client client : clients) {
                client.close().subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable disposable) {

                    }

                    @Override
                    public void onNext(Object o) {

                    }

                    @Override
                    public void onComplete() {
                        subscriber.onNext(client);
                        if (atomicInteger.addAndGet(1) == size) {
                            subscriber.onComplete();
                        }
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        subscriber.onError(throwable);
                    }
                });
            }
        });
    }


}
